﻿using Microsoft.Media.ISO;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
#if NETFX_CORE
using System.Runtime.InteropServices.WindowsRuntime;
using Windows.Foundation;
#else
#endif

namespace Microsoft.Media.IngestClient
{
    /// <summary>
    /// A class responsible for streaming data to an output consumer. This class does not contain any AMS specific functionality or business logic.
    /// </summary>
    public sealed class PushClient : IDisposable
    {
        readonly IOStreamManager streamManager = new IOStreamManager();
        readonly Queue<BoxHistory> boxes = new Queue<BoxHistory>();
        private readonly AsyncLock asyncLock = new AsyncLock();

        /// <summary>
        /// Gets the total bytes that have been provided to this object. Note: bytes may still be in buffer.
        /// </summary>
        public ulong BytesPushed { get { return (ulong)streamManager.OutputStream.Length; } }

        /// <summary>
        /// Gets the total bytes read by the output consumer.
        /// </summary>
        public ulong BytesRead { get { return (ulong)streamManager.InputStream.Length; } }

        /// <summary>
        /// Gets the total bytes successfully sent by the output consumer.
        /// </summary>
        public ulong BytesSent { get; private set; }

#if NETFX_CORE
        /// <summary>
        /// Starts a long running session that will not return until all data and future data has been sent or an error occurs.
        /// </summary>
        /// <param name="outputConsumer">The ouput consumer to send data to.</param>
        /// <returns>An awaitable action for a long running action with the ability to report on the operation progress.</returns>
        public IAsyncActionWithProgress<BoxHistory> SendBoxesAsync(IPushOutputConsumer outputConsumer)
        {
            return AsyncInfo.Run((CancellationToken c, IProgress<BoxHistory> p) => SendBoxesAsync(outputConsumer, c, p));
        }

        internal async Task SendBoxesAsync(IPushOutputConsumer outputConsumer, CancellationToken c, IProgress<BoxHistory> p)
        {
            var inputStream = new SequentialInputStream(streamManager.InputStream);
#else
        /// <summary>
        /// Starts a long running session that will not return until all data and future data has been sent or an error occurs.
        /// </summary>
        /// <param name="outputConsumer">The ouput consumer to send data to.</param>
        /// <param name="c">A cancellation token used to cancel the operation.</param>
        /// <param name="p">An object that can report on the progress of the operation.</param>
        /// <returns>An awaitable action for a long running action.</returns>
        public async Task SendBoxesAsync(IPushOutputConsumer outputConsumer, CancellationToken c, IProgress<BoxHistory> p)
        {
            var inputStream = streamManager.InputStream;
#endif

#if NETFX_CORE
            EventHandler<ulong> progressChangedHandler = (s, e) =>
#else
            ProgressEventHandler<ulong> progressChangedHandler = (s, e) =>
#endif
            {
                BytesSent = e;
                lock (boxes)
                {
                    // compare end of box position with total bytes sent to know when each box has been successfully sent.
                    while (boxes.Any() && boxes.Peek().Position <= e)
                    {
                        p.Report(boxes.Dequeue());
                    }
                }
            };
            var progress = new Progress<ulong>();
            progress.ProgressChanged += progressChangedHandler;
            try
            {
#if NETFX_CORE
                await outputConsumer.SendDataAsync(inputStream).AsTask(c, progress);
#else
                await outputConsumer.SendDataAsync(inputStream, c, progress);
#endif
            }
            finally
            {
                progress.ProgressChanged -= progressChangedHandler;
            }
        }

#if NETFX_CORE
        /// <summary>
        /// Adds a collection of boxes to the queue. Note: this is not thread-safe, PushBoxes should be used instead when calling from a background thread.
        /// </summary>
        /// <param name="boxes">The boxes to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        /// <returns>An awaitable operation that returns when the data is in the queue.</returns>
        public IAsyncAction PushBoxesAsync(IList<IBox> boxes, object userState)
        {
            return AsyncInfo.Run(c => PushBoxesAsync(boxes, userState, c));
        }

        internal async Task PushBoxesAsync(IList<IBox> boxes, object userState, CancellationToken c)
#else
        /// <summary>
        /// Adds a collection of boxes to the queue. Note: this is not thread-safe, PushBoxes should be used instead when calling from a background thread.
        /// </summary>
        /// <param name="boxes">The boxes to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        /// <param name="c">A cancellation token used to cancel the operation.</param>
        /// <returns>An awaitable operation that returns when the data is in the queue.</returns>
        public async Task PushBoxesAsync(IList<IBox> boxes, object userState, CancellationToken c)
#endif
        {
            using (await asyncLock.LockAsync(c))
            {
                foreach (var box in boxes)
                {
#if NETFX_CORE
                    await Task.Run(() => PushBox(box, userState), c);
#else
                    await TaskEx.Run(() => PushBox(box, userState), c);
#endif
                }
            }
        }

        /// <summary>
        /// Adds a collection of boxes to the queue. This is thread-safe and should be used instead of PushBoxesAsync when calling from a background thread.
        /// </summary>
        /// <param name="boxes">The boxes to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        public void PushBoxes(IList<IBox> boxes, object userState)
        {
            foreach (var box in boxes)
            {
                PushBox(box, userState);
            }
        }

#if NETFX_CORE
        /// <summary>
        /// Adds a boxe to the queue. Note: this is not thread-safe, PushBox should be used instead when calling from a background thread.
        /// </summary>
        /// <param name="box">The box to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        /// <returns>An awaitable operation that returns when the data is in the queue.</returns>
        public IAsyncAction PushBoxAsync(IBox box, object userState)
        {
            return AsyncInfo.Run(c => PushBoxAsync(box, userState, c));
        }

        internal async Task PushBoxAsync(IBox box, object userState, CancellationToken c)
#else
        /// <summary>
        /// Adds a boxe to the queue. Note: this is not thread-safe, PushBox should be used instead when calling from a background thread.
        /// </summary>
        /// <param name="box">The box to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        /// <param name="c">A cancellation token used to cancel the operation.</param>
        /// <returns>An awaitable operation that returns when the data is in the queue.</returns>
        public async Task PushBoxAsync(IBox box, object userState, CancellationToken c)
#endif
        {
            using (await asyncLock.LockAsync(c))
            {
#if NETFX_CORE
                await Task.Run(() => PushBox(box, userState), c);
#else
                await TaskEx.Run(() => PushBox(box, userState), c);
#endif
            }
        }

        /// <summary>
        /// Adds a box to the queue. This is thread-safe and should be used instead of PushBoxAsync when calling from a background thread.
        /// </summary>
        /// <param name="box">The box to queue.</param>
        /// <param name="userState">A user state that can be reported back throught the SendDataAsync propgress object.</param>
        public void PushBox(IBox box, object userState)
        {
#if NETFX_CORE
            var outputStream = streamManager.OutputStream.AsOutputStream();
#else
            var outputStream = streamManager.OutputStream;
#endif
            MP4BoxWriter.WriteBox(box, outputStream);
            streamManager.OutputStream.Flush();
            lock (boxes)
            {
                boxes.Enqueue(new BoxHistory(BytesPushed, box, userState));
            }
        }

        /// <summary>
        /// Call when complete. This signals the output consumer that the queue will not grow anymore.
        /// </summary>
        public void PushComplete()
        {
            streamManager.SignalComplete();
        }

        /// <inheritdoc />
        public void Dispose()
        {
            streamManager.Dispose();
        }
    }

    /// <summary>
    /// Represents a log of all boxes pushed to the PushClient. This object can be used to monitor progress.
    /// </summary>
    public sealed class BoxHistory
    {
        internal BoxHistory(ulong position, IBox box, object userState)
        {
            Position = position;
            Box = box;
            UserState = userState;
        }

        /// <summary>
        /// Gets the absolute position in the stream (in bytes) that the box exists.
        /// </summary>
        public ulong Position { get; private set; }

        /// <summary>
        /// Gets the box that was pushed.
        /// </summary>
        public IBox Box { get; private set; }

        /// <summary>
        /// Gets user state associated with the box when that box was pushed.
        /// </summary>
        public object UserState { get; private set; }
    }
}
